package cnooc.energy.canal;


import cn.hutool.core.util.ArrayUtil;
import cn.hutool.core.util.StrUtil;

import cnooc.common.core.redis.RedisCache;
import cnooc.energy.domain.Equipment;
import cnooc.energy.service.IDeviceDataLocationService;
import cnooc.energy.service.IEquipmentService;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.net.InetSocketAddress;
import java.util.Iterator;
import java.util.List;
import java.util.Set;

@Slf4j
@Component
public class CanalClient implements ApplicationRunner {

    @Resource
    private RedisCache redisCache;
    @Resource
    private IEquipmentService equipmentService;
    @Resource
    private IDeviceDataLocationService deviceDataLocationService;
    private static final String[] TABLE_NAMES = {"ems_equipment", "ems_device_data_location"};
    private static final String CODE_KEY = "code";
    private static final String VALUE_KEY = "table_name";
    private static final String SEPARATOR = ":";
    private static final String CANAL_SERVER_HOST = "192.168.100.177";
    private static final int CANAL_SERVER_PORT = 11111;
    private static final String CANAL_INSTANCE = "example";
    private static final String USERNAME = "";
    private static final String PASSWORD = "";

    @Value("${ems.canalEnable}")
    boolean canalEnable;


    public void initCanal() {
        // 创建链接
        CanalConnector connector = CanalConnectors.newSingleConnector(
                new InetSocketAddress(CANAL_SERVER_HOST, CANAL_SERVER_PORT),
                CANAL_INSTANCE, USERNAME, PASSWORD);
        int batchSize = 1000;
        try {
            log.info("启动 canal 数据同步...") ;
            connector.connect();
            connector.subscribe(".*\\..*");
            connector.rollback();
            while (true) {
                // 获取指定数量的数据
                Message message = connector.getWithoutAck(batchSize);
                long batchId = message.getId();
                int size = message.getEntries().size();

                if (batchId == -1 || size == 0) {
                    try {
                        // 时间间隔1000毫秒
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                } else {

//                    syncEntry(message.getEntries());
                }
                connector.ack(batchId); // 提交确认
                // connector.rollback(batchId); // 处理失败, 回滚数据
            }
        } finally {
            connector.disconnect();
        }
    }

    private void syncEntry(List<CanalEntry.Entry> entrys) {
        for (CanalEntry.Entry entry : entrys) {
            if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN
                    || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
                continue;
            }

            CanalEntry.RowChange rowChange;
            try {
                rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
            } catch (Exception e) {
                throw new RuntimeException("ERROR data:" + entry.toString(), e);
            }

            CanalEntry.EventType eventType = rowChange.getEventType();
            log.info("================> binlog[{}:{}] , name[{},{}] , eventType : {}",
                    entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
                    entry.getHeader().getSchemaName(), entry.getHeader().getTableName(),
                    eventType);

            String tableName = entry.getHeader().getTableName();
            if (!ArrayUtil.contains(TABLE_NAMES, tableName)) continue;

            for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
                if (eventType == CanalEntry.EventType.INSERT || eventType == CanalEntry.EventType.UPDATE) {
                    redisUpdateOrInsert(tableName, rowData.getAfterColumnsList());
                } else if (eventType == CanalEntry.EventType.DELETE) {
                    redisDelete(tableName, rowData.getBeforeColumnsList());
                }
            }
        }
    }


    private void redisUpdateOrInsert(String tableName, List<CanalEntry.Column> columns) {
        if (tableName.equals(TABLE_NAMES[0])) {
            String key = null;
            String value = null;
            Iterator<CanalEntry.Column> iterator = columns.iterator();
            while (StrUtil.isBlank(key) || StrUtil.isBlank(value)) {
                CanalEntry.Column next = iterator.next();
                if (next.getName().equals(VALUE_KEY)) {
                    value = next.getValue();
                    continue;
                }
                if (next.getName().equals(CODE_KEY)) {
                    key = next.getValue();
                }
            }
            redisCache.setCacheObject(tableName + SEPARATOR + key + SEPARATOR + "tableName", value);
        } else {
            for (CanalEntry.Column next : columns) {
                if (next.getName().equals("device_id")&StrUtil.isNotBlank(next.getValue())) {
                    Equipment equipment = equipmentService.selectEquipmentById(Long.valueOf(next.getValue()));

                    Set<String> ddlList = deviceDataLocationService.locationByDeviceId(equipment.getId().toString());
                    String locationKey = TABLE_NAMES[0] + SEPARATOR + equipment.getCode() + SEPARATOR + "locations";
                    if (redisCache.hasKey(locationKey)) {
                        redisCache.deleteObject(locationKey);
                    }
                    redisCache.setCacheSet(locationKey, ddlList);

                    break;


                }
            }
        }

    }

    private void redisDelete(String tableName, List<CanalEntry.Column> columns) {
        if (tableName.equals(TABLE_NAMES[0])) {
            for (CanalEntry.Column column : columns) {
                if (CODE_KEY.equalsIgnoreCase(column.getName())) {
                    String key = tableName + SEPARATOR + column.getValue() + SEPARATOR + "tableName";
                    redisCache.deleteObject(key);
                    log.info("redis数据同步删除，key：" + key);
                    break;
                }
            }
        } else {
            for (CanalEntry.Column next : columns) {
                if (next.getName().equals("device_id")) {
                    Equipment equipment = equipmentService.selectEquipmentById(Long.valueOf(next.getValue()));
                    Set<String> ddlList = deviceDataLocationService.locationByDeviceId(equipment.getId().toString());
                    String locationKey = TABLE_NAMES[0] + SEPARATOR + equipment.getCode() + SEPARATOR + "locations";
                    redisCache.deleteObject(locationKey);
                    redisCache.setCacheSet(locationKey, ddlList);
                    break;
                }
            }
        }

    }


    @Override
    public void run(ApplicationArguments args) throws Exception {
            log.info("--------同步设备码以及对应数据表-------");
            equipmentService.selectEquipmentList(new Equipment()).forEach(equipment -> {
                String key = TABLE_NAMES[0] + SEPARATOR + equipment.getCode() + SEPARATOR + "tableName";

                if (redisCache.hasKey(key)) {
                    redisCache.deleteObject(key);
                }

                if (StrUtil.isNotBlank(equipment.getTableName())) {
                    Set<String> ddlList = deviceDataLocationService.locationByDeviceId(equipment.getId().toString());
                    redisCache.setCacheObject(key, equipment.getTableName());
                    log.info("已同步设备编号为:{},设备名称为:{},对应数据表为:{}", equipment.getCode(), equipment.getName(), equipment.getTableName());
                    if (!ddlList.isEmpty()) {
                        String locationKey = TABLE_NAMES[0] + SEPARATOR + equipment.getCode() + SEPARATOR + "locations";
                        if (redisCache.hasKey(locationKey)) {
                            redisCache.deleteObject(locationKey);
                        }
                        redisCache.setCacheSet(locationKey, ddlList);
                    }
                }
            });
//            this.initCanal();
        }

}